package com.whq.thrift.api.web.admin.controller.bs.test;

import java.util.Properties;

import javax.servlet.http.HttpServletRequest;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;

import com.whq.common.StringUtil;
import com.whq.thrift.api.common.kafka.KafkaStringProducer;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

@Controller
@Scope("prototype")
public class KafkaTestController {
	private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStringProducer.class);

	@RequestMapping(value = "bs/test/kafkatest/index")
	public String index(@RequestParam(value = "ipAddress", required = false) String ipAddress, @RequestParam(value = "port", required = false) Integer port,
			@RequestParam(value = "topic", required = false) String topic, @RequestParam(value = "key", required = false) String key,
			@RequestParam(value = "value", required = false) String value, Model model, HttpServletRequest request) {
		model.addAttribute("ipAddress", ipAddress);
		model.addAttribute("port", port);
		model.addAttribute("topic", topic);
		model.addAttribute("key", key);
		model.addAttribute("value", value);

		if (StringUtil.isBlank(key)) {
			key = System.currentTimeMillis() + "";
		}

		String val = "";
		if (StringUtil.isBlank(ipAddress) || (port == null) || StringUtil.isBlank(topic) || StringUtil.isBlank(value)) {
			model.addAttribute("val", val);
			return "bs/test/kafkatest/index";
		}
		try {
			Properties props = new Properties();
			// 此处配置的是kafka的端口
			props.put("metadata.broker.list", ipAddress+":"+port);
			// 配置value的序列化类
			props.put("serializer.class", "kafka.serializer.StringEncoder");
			// 配置key的序列化类
			props.put("key.serializer.class", "kafka.serializer.StringEncoder");
			props.put("request.required.acks", "-1");
			ProducerConfig config = new ProducerConfig(props);
			Producer<String, String> producer = new Producer<String, String>(config);

			if (producer != null) {
				producer.send(new KeyedMessage<String, String>(topic, key, value));
			}

			if (producer != null) {
				producer.close();
			}
		} catch (Exception e) {
			model.addAttribute("val", "producer exception:" + e);
			LOGGER.error("close kafka producer exception: ", e);
			return "bs/test/kafkatest/index";
		}
		model.addAttribute("val", "succuss");

		return "bs/test/kafkatest/index";
	}


}
